package org.example.datasource;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import org.example.datasource.eventhandler.AbstractEventHandler;
import org.example.datasource.eventhandler.EventHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.kafka.consumer.CanalKafkaConsumer;
import com.google.common.collect.Maps;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class DataSourceHandler {
    
    @Autowired
    private CanalKafkaConsumer canalKafkaConsumer;
    
    @PostConstruct
    public void loadData() {
        AbstractEventHandler.check();
        log.info("============loadData===========");
        new Thread(()->{
            while (true) {
                List<CommonMessage> msgs = canalKafkaConsumer.getMessage(30l, TimeUnit.SECONDS);
                if(msgs!=null && !msgs.isEmpty()) {
                    for(CommonMessage cm : msgs) {
                        AbstractEventHandler.getHandler(cm.getType()).hand(cm);
                    }
                }else {
                    log.info("===============等待接受数据===================");
                }
            }
        }).start();
    }
}
